package com.wyclabs.mq.eventbus;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;

/**
 * @author TylerW
 * @date 2020/4/7 13:47
 */
@Slf4j
public class DefaultBusEventManager implements EventManager, InitializingBean, ApplicationContextAware {

    private static final String EVENT_HANDLE_METHOD_NAME = "handleEvent";

    private AsyncTaskExecutor asyncTaskExecutor;

    private RabbitTemplate rabbitTemplate;

    private String topic;

    private String uniqueRoutingKey;

    private String sharedRoutingKey;

    private String uniqueRoutingKeyTemplate;

    private ApplicationContext applicationContext;

    private List<EventMapping> mappings = Lists.newArrayList();

    public DefaultBusEventManager(RabbitTemplate rabbitTemplate, String topic, String uniqueRoutingKey,
                                  String sharedRoutingKey, String uniqueRoutingKeyTemplate) {
        super();
        this.rabbitTemplate = rabbitTemplate;
        this.topic = topic;
        this.uniqueRoutingKey = uniqueRoutingKey;
        this.sharedRoutingKey = sharedRoutingKey;
        this.uniqueRoutingKeyTemplate = uniqueRoutingKeyTemplate;
    }

    @Override
    public void afterPropertiesSet() {
        Collection<EventHandler> handlers = applicationContext.getBeansOfType(EventHandler.class)
                .values();

        if (CollectionUtils.isEmpty(handlers)) {
            return;
        }

        for (EventHandler handler : handlers) {
            Class clazz = handler.getClass();
            for (Method method : clazz.getMethods()) {
                // @formatter:off
                if (!EVENT_HANDLE_METHOD_NAME.equals(method.getName())
                        && Modifier.isPublic(method.getModifiers())) {
                    continue;
                }
                // @formatter:on

                handlerMethod(method, handler);
            }

        }
    }

    protected void handlerMethod(Method method, EventHandler handler) {
        Class<?>[] parameters = method.getParameterTypes();
        if (parameters.length == 1) {
            Class<?> parameterType = parameters[0];
            boolean added = false;

            if (!Event.class.isAssignableFrom(parameterType)) {
                return;
            }

            for (EventMapping mapping : mappings) {
                if (mapping.clazz.equals(parameterType)) {
                    added = mapping.handlers.add(handler);
                    break;
                }
            }

            if (!added) {
                EventMapping mapping = new EventMapping(parameterType);
                mapping.handlers.add(handler);
                mappings.add(mapping);
            }
        }

    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }

    /**
     * 发布事件
     * @param event 事件
     * @author Zhang WenTao
     * @version 1.0 create at 2019年4月10日
     */
    public void publish(Event event, PublishStrategy strategy) {
        Assert.notNull(event, "the event must not be null!!");
        Assert.notNull(strategy, "the strategy must not be null!!");

        switch (strategy) {
            case LOCAL:
                publishLocalAndIgnoreException(event);
                break;
            case ASYNC:
                asyncTaskExecutor.execute(() -> publishLocalAndIgnoreException(event), 0);
                break;
            case RABBIT_MQ:
            case OVERALL:
                String tp = DefaultBusEventManager.this.topic;
                // @formatter:off
                String routingKey = PublishStrategy.RABBIT_MQ.equals(strategy)
                        ? DefaultBusEventManager.this.uniqueRoutingKey
                        : DefaultBusEventManager.this.sharedRoutingKey;
                // @formatter:on
                publishRabbitEvent(event, tp, routingKey);
                break;
            default:
                throw new UnsupportedOperationException(strategy.name());
        }
    }

    @Override
    public void publish(DirectionalEvent event) {
        Assert.notNull(event, "the event must not be null!!");
        String[] servers = event.servers();

        Assert.notEmpty(servers, "the servers must not be empty!!");

        final String tp = DefaultBusEventManager.this.topic;
        final String urkt = DefaultBusEventManager.this.uniqueRoutingKeyTemplate;
        for (String server : servers) {
            publishRabbitEvent(event, tp, String.format(urkt, server));
        }

    }

    /**
     * 发布Rabbit事件
     */
    protected void publishRabbitEvent(Event event, String topic, String routingKey) {
        try {
            if (event instanceof ConsistenceEvent) {
                //如果支持发布确认
                rabbitTemplate.convertAndSend(topic, routingKey, event, ((ConsistenceEvent) event).getCorrelationData());
            } else {
                rabbitTemplate.convertAndSend(topic, routingKey, event);
            }
        } catch (Exception e) {
            log.error(String.format("发布事件到RabbitMQ失败 routingKey: %s", routingKey), e);
        }
    }

    /**
     * 发布事件且忽略所有异常
     */
    private void publishLocalAndIgnoreException(Event event) {
        publishLocalEvent(event, true);
    }

    /**
     * 发布本地事件
     */
    protected void publishLocalEvent(Event event, boolean ignoreException) {
        Class clazz = event.getClass();
        final List<EventMapping> ms = this.mappings;
        for (EventMapping mapping : ms) {
            if (!clazz.isAssignableFrom(mapping.clazz)) {
                continue;
            }

            Iterator<EventHandler> itr = mapping.iterator();
            int count = 0;
            while (itr.hasNext()) {
                try {
                    itr.next().handleEvent(event);
                    count++;
                } catch (Exception e) {
                    if (ignoreException) {
                        log.error("事件处理失败 event: " + JSON.toJSONString(event), e);
                    } else {
                        log.error("事件处理失败", e);
                        throw new EventHandleException("事件处理失败", e);
                    }
                }
            }

            if (count == 0) {
                log.warn("事件成功接收, 但没有可用处理器进行处理");
            }
        }

    }

    @RabbitListener(queues = EventBusAutoConfiguration.RabbitEventBusConfiguration.QUEUE_NAME_EL)
    protected class RabbitEventListener {

        @RabbitHandler
        public void handleEvent(@Payload Event event) {
            publishLocalEvent(event, false);
        }

    }

    /**
     * EventMapping
     * @param <T>
     * @author Zhang WenTao
     * @version 1.0 create at 2019年4月10日
     */
    @RequiredArgsConstructor
    private static class EventMapping<T extends Event> implements Iterable<EventHandler<T>> {

        @NonNull
        private final Class<T> clazz;

        private final List<EventHandler<T>> handlers = Lists.newArrayList();

        @Override
        public Iterator<EventHandler<T>> iterator() {
            return handlers.iterator();
        }

    }

}
